from collections import defaultdict
from typing import Dict, List, Optional, Tuple

from cold_start_analyzer import db_helper


def _fetch_callstack_records(
    cursor,
    thread_name: str,
    start_ts: Optional[int],
    end_ts: Optional[int],
) -> List[Dict[str, object]]:
    normalized = db_helper._normalize_thread_name(thread_name)
    params: List[object] = []
    query = """
    SELECT callstack.id, callstack.parent_id, callstack.name, callstack.ts, callstack.dur
    FROM callstack
    JOIN thread ON callstack.callid = thread.itid
    WHERE thread.name LIKE ?
    """
    params.append(f"%{normalized}%") if normalized else params.append("%")
    if start_ts is not None:
        query += " AND callstack.ts >= ?"
        params.append(start_ts)
    if end_ts is not None:
        query += " AND callstack.ts <= ?"
        params.append(end_ts)
    query += " ORDER BY callstack.ts"
    cursor.execute(query, tuple(params))
    columns = ["id", "parent_id", "name", "ts", "dur"]
    return [dict(zip(columns, row)) for row in cursor.fetchall()]


def _aggregate_hotspots(
    records: List[Dict[str, object]],
    top_n: int,
    weights: Optional[Dict[int, float]] = None,
) -> List[Dict[str, object]]:
    if top_n <= 0:
        return []
    totals: Dict[str, float] = defaultdict(float)
    for record in records:
        name = record.get("name") or ""
        if not name:
            continue
        if weights is not None:
            dur = float(weights.get(record["id"], 0.0))
        else:
            dur = float(record.get("dur") or 0)
        totals[name] += dur
    sorted_items = sorted(totals.items(), key=lambda item: item[1], reverse=True)
    total_duration = sum(d for _, d in sorted_items) or 1
    hotspots: List[Dict[str, object]] = []
    for name, duration in sorted_items[:top_n]:
        hotspots.append(
            {
                "name": name,
                "cost": duration,
                "percent": (duration / total_duration) * 100,
            }
        )
    return hotspots


def build_callstack_table(
    cursor,
    thread_name: str,
    start_ts: Optional[int],
    end_ts: Optional[int],
    depth: int,
    max_rows: int,
    hotspot_count: int,
    min_ratio: float = 0.05,
    total_duration_override: Optional[float] = None,
) -> Tuple[List[Dict[str, object]], Tuple[Optional[int], Optional[int]], List[Dict[str, object]], List[Dict[str, object]]]:
    if depth <= 0 or max_rows <= 0:
        return [], (start_ts, end_ts), [], []
    records = _fetch_callstack_records(cursor, thread_name, start_ts, end_ts)
    if not records:
        return [], (start_ts, end_ts), [], []

    actual_start = records[0]["ts"]
    actual_end = max((record["ts"] + (record["dur"] or 0)) for record in records)
    nodes = {record["id"]: record for record in records}
    children: Dict[Optional[int], List[int]] = defaultdict(list)
    for record in records:
        parent_id = record["parent_id"]
        if parent_id in nodes:
            children[parent_id].append(record["id"])
        else:
            children[None].append(record["id"])

    rows: List[Dict[str, object]] = []
    computed_total = sum(record.get("dur", 0) or 0 for record in records) or 0
    total_duration = float(total_duration_override) if total_duration_override else computed_total

    def add_row(name: str, level: int, cost: float, is_other: bool = False) -> bool:
        if len(rows) >= max_rows:
            return False
        percent = (cost / total_duration * 100) if total_duration else 0
        rows.append(
            {
                "level": level,
                "name": name,
                "cost": cost,
                "percent": percent,
                "is_other": is_other,
            }
        )
        return True

    def walk(node_id: int, level: int, parent_cost: float) -> bool:
        if len(rows) >= max_rows:
            return False
        record = nodes[node_id]
        cost = float(record.get("dur") or 0)
        label = record.get("name") or ""
        if not add_row(label, level, cost):
            return False
        if level >= depth or cost <= 0:
            return True
        other_cost = 0.0
        child_ids = children.get(node_id, [])
        child_ids.sort(key=lambda cid: float(nodes[cid].get("dur") or 0), reverse=True)
        for idx, child_id in enumerate(child_ids):
            child_cost = float(nodes[child_id].get("dur") or 0)
            ratio = (child_cost / cost) if cost else 0
            if idx >= 5 or level >= depth - 1 or (level >= 1 and ratio < min_ratio):
                other_cost += child_cost
                continue
            if not walk(child_id, level + 1, cost):
                return False
        if other_cost > 0 and len(rows) < max_rows:
            add_row(
                f"其他(<{int(min_ratio * 100)}%)",
                level + 1,
                other_cost,
                is_other=True,
            )
        return True

    if total_duration > 0:
        root_children = children.get(None, [])
        root_children.sort(key=lambda cid: float(nodes[cid].get("dur") or 0), reverse=True)
        other_root_cost = 0.0
        for idx, node_id in enumerate(root_children):
            if idx >= 5:
                other_root_cost += float(nodes[node_id].get("dur") or 0)
                continue
            if len(rows) >= max_rows:
                break
            walk(node_id, 1, total_duration)
        if other_root_cost > 0 and len(rows) < max_rows:
            add_row(
                f"其他(<{int(min_ratio * 100)}%)",
                1,
                other_root_cost,
                is_other=True,
            )

    self_costs: Dict[int, float] = {}
    for record in records:
        node_id = record["id"]
        child_sum = sum(float(nodes[child_id].get("dur") or 0) for child_id in children.get(node_id, []))
        self_cost = float(record.get("dur") or 0) - child_sum
        if self_cost < 0:
            self_cost = 0.0
        self_costs[node_id] = self_cost

    hotspots_wall = _aggregate_hotspots(records, top_n=hotspot_count)
    hotspots_self = _aggregate_hotspots(records, top_n=hotspot_count, weights=self_costs)

    return rows, (actual_start, actual_end), hotspots_wall, hotspots_self
